package cn.chengpengper.seckill.order.application.place.impl;

import cn.chengpengper.dubbo.interfaces.goods.SeckillGoodsDubboService;
import cn.chengpengper.seckill.common.cache.distribute.DistributedCacheService;
import cn.chengpengper.seckill.common.code.HttpCode;
import cn.chengpengper.seckill.common.constants.SeckillConstants;
import cn.chengpengper.seckill.common.exception.SeckillException;
import cn.chengpengper.seckill.common.model.dto.SeckillGoodsDTO;
import cn.chengpengper.seckill.common.model.message.TxMessage;
import cn.chengpengper.seckill.common.utils.id.IdUtils;
import cn.chengpengper.seckill.mq.MessageSenderService;
import cn.chengpengper.seckill.order.application.model.command.SeckillOrderCommand;
import cn.chengpengper.seckill.order.application.place.SeckillPlaceOrderService;
import cn.chengpengper.seckill.order.domain.model.entity.SeckillOrder;
import cn.chengpengper.seckill.order.domain.service.SeckillOrderDomainService;
import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author chengpeng.hu
 */
@Slf4j
@Service
@ConditionalOnProperty(name = "place.order.type", havingValue = "lua")
public class SeckillPlaceOrderLuaServiceImpl implements SeckillPlaceOrderService {

    @Autowired
    private SeckillOrderDomainService seckillOrderDomainService;

    @DubboReference(version = "1.0.0")
    private SeckillGoodsDubboService seckillGoodsDubboService;

    @Autowired
    private DistributedCacheService distributedCacheService;

    @Autowired
    private MessageSenderService messageSenderService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void saveOrderInTransaction(TxMessage txMessage) {
        try {
            Boolean submitTransaction = distributedCacheService.hasKey(SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo())));
            if (Boolean.TRUE.equals(submitTransaction)) {
                log.info("saveOrderInTransaction|已经执行过本地事务|{}", txMessage.getTxNo());
                return;
            }
            //构建订单
            SeckillOrder seckillOrder = this.buildSeckillOrder(txMessage);
            //保存订单
            seckillOrderDomainService.saveSeckillOrder(seckillOrder);
            //保存事务日志
            distributedCacheService.put(SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo())), txMessage.getTxNo(), SeckillConstants.TX_LOG_EXPIRE_DAY, TimeUnit.DAYS);
        } catch (Exception e) {
            log.error("saveOrderInTransaction|异常|{}", e.getMessage());
            // 出异常，回滚库存，删除事务标志
            distributedCacheService.delete(SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo())));
            this.rollbackCacheStock(txMessage);
            throw e;
        }
    }

    /**
     * 回滚缓存库存
     */
    private void rollbackCacheStock(TxMessage txMessage) {
        log.info("SeckillPlaceOrderLuaService|rollbackCacheStack|txNo:{},exceptionFlag:{}", txMessage.getTxNo(), txMessage.getException());
        //扣减过缓存库存
        if (Boolean.FALSE.equals(txMessage.getException())) {
            String luaKey = SeckillConstants.getKey(SeckillConstants.ORDER_TX_KEY, String.valueOf(txMessage.getTxNo())).concat(SeckillConstants.LUA_SUFFIX);
            Long result = distributedCacheService.checkExecute(luaKey, SeckillConstants.TX_LOG_EXPIRE_SECONDS);
            //已经执行过恢复缓存库存的方法
            if (NumberUtil.equals(result, SeckillConstants.ALREADY_EXECUTE)) {
                log.info("handlerCacheStock|已经执行过恢复缓存库存的方法|{}", JSONObject.toJSONString(txMessage));
                return;
            }
            //只有分布式锁方式和Lua脚本方法才会扣减缓存中的库存
            String key = SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_STOCK_KEY_PREFIX, String.valueOf(txMessage.getGoodsId()));
            distributedCacheService.incrementByLua(key, txMessage.getQuantity());
        }
    }

    @Override
    public Long placeOrder(Long userId, SeckillOrderCommand seckillOrderCommand) {
        long txNo = IdUtils.getNextId();
        boolean exceptionFlag = false;
        Long decrementResult = 0L;
        SeckillGoodsDTO seckillGoods = seckillGoodsDubboService.getSeckillGoods(seckillOrderCommand.getGoodsId(), seckillOrderCommand.getVersion());
        checkSeckillOrder(seckillOrderCommand, seckillGoods);
        String key = SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_STOCK_KEY_PREFIX, String.valueOf(seckillGoods.getId()));
        try {
            Object limitObj = distributedCacheService.getObject(SeckillConstants.getKey(SeckillConstants.GOODS_ITEM_LIMIT_KEY_PREFIX, String.valueOf(seckillOrderCommand.getGoodsId())));
            if (Objects.isNull(limitObj)) {
                throw new SeckillException(HttpCode.GOODS_OFFLINE);
            }
            if (Integer.parseInt(String.valueOf(limitObj)) < seckillOrderCommand.getQuantity()) {
                throw new SeckillException(HttpCode.BEYOND_LIMIT_NUM);
            }
            decrementResult = distributedCacheService.decrementByLua(key, seckillOrderCommand.getQuantity());
            distributedCacheService.checkResult(decrementResult);
            log.info("SeckillPlaceOrderLuaService|扣减缓存库存成功|txNo:{},quantity:{}", txNo, seckillOrderCommand.getQuantity());
        } catch (Exception e) {
            exceptionFlag = true;
            // 将内存中的库存增加回去
            if (decrementResult == SeckillConstants.LUA_RESULT_EXECUTE_TOKEN_SUCCESS) {
                log.error("SeckillPlaceOrderLockServiceImpl|下单发生异常，缓存中库存已扣除，增加回来");
                distributedCacheService.incrementByLua(key, seckillOrderCommand.getQuantity());
            }
        }
        //发送事务消息
        messageSenderService.sendMessageInTransaction(this.getTxMessage(SeckillConstants.TOPIC_TX_MSG, txNo, userId, SeckillConstants.PLACE_ORDER_TYPE_LUA, exceptionFlag, seckillOrderCommand, seckillGoods, 0,seckillOrderCommand.getOrderTaskId()), null);
        return txNo;
    }
}
